package cn.linkai.hadoop.storm;

import org.apache.storm.Config;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.topology.TopologyBuilder;

public class TopoMain {

    public static void main(String[] args) {
        TopologyBuilder builder=new TopologyBuilder();

        builder.setSpout("randomspout",new RandomWordSpout(),4);

        builder.setBolt("upperbolt",new UpperBolt(),4).shuffleGrouping("randomspout");

        builder.setBolt("suffixbolt",new SuffixBolt(),4).shuffleGrouping("upperbolt");

        StormTopology demotop=builder.createTopology();

        Config config=new Config();

        config.setNumWorkers(4);
        config.setDebug(true);
        config.setNumAckers(0);

        try {
            StormSubmitter.submitTopology("demotop",config,demotop);
        } catch (AlreadyAliveException e) {
            e.printStackTrace();
        } catch (InvalidTopologyException e) {
            e.printStackTrace();
        } catch (AuthorizationException e) {
            e.printStackTrace();
        }
    }
}
